package cn.itcast.streaming.function.watermark;

import cn.itcast.streaming.bean.ItcastDataObj;
import cn.itcast.streaming.utils.DateUtil;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

//驾驶行程自定义水印对象，解决数据延迟30s问题
public class TripDriveWaterMark implements AssignerWithPeriodicWatermarks<ItcastDataObj> {

    //当前窗口最大水印时间
    private long currentMaxTimestamp = 0L;
    //允许最大的乱序时间30S
    private final long maxOutofOrderness = 30*1000L;

    @Nullable
    @Override //生成当前水印
    public Watermark getCurrentWatermark() {
        //System.out.println("水印时间="+(currentMaxTimestamp - maxOutofOrderness));
        return new Watermark(currentMaxTimestamp - maxOutofOrderness);
    }

    @Override
    public long extractTimestamp(ItcastDataObj itcastDataObj, long l) {
        currentMaxTimestamp = Math.max(itcastDataObj.getTerminalTimeStamp(),currentMaxTimestamp);
        System.out.println("当前数据时间："+ DateUtil.timestampToDataStr(itcastDataObj.getTerminalTimeStamp())+" 窗口最大时间："+DateUtil.timestampToDataStr(currentMaxTimestamp));
        return currentMaxTimestamp;
    }
}
